package com.atguigu.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by undeRdoG on  2021-06-08  14:26
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object Spark18_RDD_Operator_Transform {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("Operator").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)


    val rdd = sc.makeRDD(List(("a", 1), ("2", 2), ("b", 3), ("b", 4),("b",5),("a",6)),2)


    /**
    *   aggregateByKey最终的返回值的类型，是取决于 初始值的类型
    * */

    // 获取相同key的平均值   (a,3)   (b,4)
    val res: RDD[(String, (Int, Int))] = rdd.aggregateByKey((0, 0))(
      (t, v) => {
        (t._1 + v, t._2 + 1)
      },
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    )
    val result: RDD[(String, Int)] = res.mapValues {
      case (num, cnt) => {
        num / cnt
      }
    }
    result.collect().foreach(println)
  }
}